#include "DeviceRecordCache.h"
#include "mpeg-ps.h"
#include "mpeg-ts-proto.h"
#include "jsoncpp/assertions.h"
#include "Extension/H264.h"
#include "Util/System.h"
#include <execinfo.h>
#if defined(NDEBUG)
#define MAX_CACHE_IDLE_TIME 3600
#else
#define MAX_CACHE_IDLE_TIME 120
#endif
////////////////////////////////////////////CacheSubScriber订阅者实现/////////////////////////////////////////////////
template<typename frame>
FrameRecordCache<frame>::CacheSubScriber::CacheSubScriber(EventPoller::Ptr poller,FrameRecordCache* pCache,const privateproto::DeviceChannel& dev_chan,time_t start_push_time,time_t end_push_time)
:m_pCacher(pCache),m_dev_chan(dev_chan) {
    m_poller = poller;
    m_start_push_time = start_push_time;
    m_end_push_time = end_push_time;
    m_playscalevalue = 1.0f;
    m_noframecounter = 0;
    m_stateType = SubscriberStateType::SST_UNKNOWN;
    m_origindesc = std::make_shared<CacheOrigin>(this);
    m_muxer = std::make_shared<MultiMediaSourceMuxer>(m_poller,dev_chan._vhost,
                            CACHE_APP_NAME,//保持原来的schema一致
                            dev_chan.GenStreamId(), 0.0f,
                            true, true, false, false);
    m_muxer->setMediaListener(m_origindesc);
}

template<typename frame>
FrameRecordCache<frame>::CacheSubScriber::~CacheSubScriber(){
    m_muxer->resetTracks();
    m_close_timer = nullptr;
    m_send_pkt_timer = nullptr;
}
template<typename frame>
void FrameRecordCache<frame>::CacheSubScriber::ControlStream(const std::string command,const double scale) {
    m_poller->async([this,command,scale]() {
            TraceL << "开始执行async, command:" << command << ",scale:" << scale;
            if(command.compare("PLAY") == 0) {
                m_stateType = SubscriberStateType::SST_PUSHING;
                m_playscalevalue = 1.0f;
            }else if(command.compare("PAUSE") == 0) {
                m_stateType = SubscriberStateType::SST_PAUSE;
                m_playscalevalue = 1.0f;
            }else if(command.compare("SCALE") == 0) {
                m_playscalevalue = (scale > 4.0f)?4.0f:scale;//只支持8倍速以下
                float send_packet_interval = 1.0f / m_pCacher->GetFps();
                send_packet_interval/=m_playscalevalue;
                WarnL << "调整定时间隔:" << send_packet_interval;
                m_send_pkt_timer->UpdateInterval(send_packet_interval);
                m_stateType = SubscriberStateType::SST_PUSHING;//调整倍速的同事取消暂停状态
            }else {
                ErrorL << "unhandled command:" << command;
            }
            TraceL << "执行后, state:" << getSubscriberStateTypeString(m_stateType) << ",scale:" << m_playscalevalue;
    });
}
template<typename frame>
Value FrameRecordCache<frame>::CacheSubScriber::DumpAsJson() {
    Value root;
    root["play_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",m_start_push_time);
    root["play_end_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",m_end_push_time);
    root["curPushPts"] = Int64(m_curPushPts);
    root["dev_chan"] = m_dev_chan.ToString();
    root["m_curPushPosInList"] = Int64(m_curPushPosInList);
    root["reader count"] = Int64(m_muxer->totalReaderCount());
    root["state_type"] = getSubscriberStateTypeString(m_stateType);
    
    return root;
}

template<typename frame>
void FrameRecordCache<frame>::CacheSubScriber::TriggerDelayedClose(){
    GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS);
      m_close_timer = std::make_shared<Timer>(stream_none_reader_delay / 1000.0f, [&]() {
        int cur_reader_count = GetReaderCount();
        if(cur_reader_count <= 0) {
            WarnL << "执行延迟关闭,counter:" << cur_reader_count;
            CloseSubscriber();
        }else{
            DebugL << "取消延迟关闭,counter:" << cur_reader_count;
        } 
        return false;
    }, nullptr);
}
#if 0
template<typename frame>
int FrameRecordCache<frame>::CacheSubScriber::StartPushRecordStream() {
    m_stateType = SubscriberStateType::SST_PUSHING;
    float send_packet_interval = 1.0f/m_pCacher->GetFps();
    printf("[%s:%d:%s]推送流,m_poller:%p,m_start_push_time:%ld,m_end_push_time:%ld,send_packet_interval:%.3f\n",__FUNCTION__,__LINE__,getTimeStr("%Y-%m-%d %H:%M:%S",time(NULL)).c_str(),m_poller.get(),m_start_push_time,m_end_push_time,send_packet_interval);    
    
    m_curPushPosInList = m_pCacher->LocateFrame(m_start_push_time,m_end_push_time);
    #if !defined(NDEBUG)
    printf("[%s:%d]m_curPushPosInList:%lu,size:%lu\n",__FUNCTION__,__LINE__,m_curPushPosInList,m_pCacher->GetFrameCount());      
    #endif
    m_send_pkt_timer = std::make_shared<Timer>(send_packet_interval,[=](){
        #if !defined(NDEBUG)
        TraceL << "定时发送数据,时间间隔:" << send_packet_interval << "帧率:" << m_pCacher->GetFps() << ",m_playscalevalue:" << m_playscalevalue;
        #endif
        if(m_stateType == SubscriberStateType::SST_PAUSE) {
            return true;
        }
        auto pframes = m_pCacher->GetFrameByRange(m_curPushPosInList,send_packet_interval*m_playscalevalue);
        m_curPushPosInList += pframes.size();
        for(auto it = pframes.begin();it != pframes.end(); it++) {
            auto pframe = *it;
            if(pframe == nullptr) {
                WarnL << m_dev_chan.ToString() << "耗尽缓存" ;
                return false;//不能再发送了
            }
            m_curPushPts = pframe->pts();
            if(m_pCacher->CheckCacheExhausted(pframe->pts())) {
                if(m_last_check_cache.elapsedTime() > 15 * 1000){//间隔十五秒以内不要重复触发缓存耗尽hook
                    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastCacheExhaust, m_dev_chan, m_pCacher->GetPlayEndTime());
                    DebugL << "触发缓存耗尽拉流事件 : " << m_dev_chan.ToString() << ",time:" << m_pCacher->GetPlayEndTime();
                    m_last_check_cache.resetTime();
                }
            }
            m_muxer->inputFrame(pframe);
        }
        //回调_onsendpkt
        return true;
    }, m_poller);
    return 0;
}
#else
template<typename frame>
int FrameRecordCache<frame>::CacheSubScriber::StartPushRecordStream() {
    m_stateType = SubscriberStateType::SST_PUSHING;
    printf("[%s:%d:%s]推送流,m_poller:%p,m_start_push_time:%ld,m_end_push_time:%ld\n",__FUNCTION__,__LINE__,getTimeStr("%Y-%m-%d %H:%M:%S",time(NULL)).c_str(),m_poller.get(),m_start_push_time,m_end_push_time);    
    float send_packet_interval = 1.0f / m_pCacher->GetFps();
    send_packet_interval/=m_playscalevalue;
    #if !defined(NDEBUG)
    printf("[%s:%d]send_packet_interval:%.3f\n",__FUNCTION__,__LINE__,send_packet_interval);
    #endif

    m_curPushPosInList = m_pCacher->LocateFrame(m_start_push_time,m_end_push_time);
    if(m_curPushPosInList < 0) {
        return m_curPushPosInList;
    }
    #if !defined(NDEBUG)
    printf("[%s:%d]m_curPushPosInList:%d,size:%lu\n",__FUNCTION__,__LINE__,m_curPushPosInList,m_pCacher->GetFrameCount());      
    #endif
    m_send_pkt_timer = std::make_shared<MutableTimer>(send_packet_interval,[&](){
        #if !defined(NDEBUG)
        //TraceL << "定时发送数据,时间间隔:" << send_packet_interval << "帧率:" << m_pCacher->GetFps();
        #endif
        if(m_stateType == SubscriberStateType::SST_PAUSE) {
            return true;
        }
        auto pframe = m_pCacher->GetFrame(m_curPushPosInList);
        if(pframe == nullptr) {
            if(m_noframecounter++*send_packet_interval > 10.0f){
                WarnL << m_dev_chan.ToString() << "耗尽缓存超时" ;
                m_stateType == SubscriberStateType::SST_CLOSED;
                return false;//不再发送
            }else{
                return true;//继续
            }
        }else{
            m_curPushPosInList++;
            m_noframecounter = 0;
            m_curPushPts = pframe->pts();
            if(!m_pCacher->IsAbnormal()&&m_pCacher->CheckCacheExhausted(pframe->pts())) {
                if(m_last_check_cache.elapsedTime() > 30 * 1000){//间隔半分钟以内不要重复触发缓存耗尽hook
                    WarnL << m_dev_chan.ToString() << "发送将要耗尽缓存事件" ;
                    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastCacheExhaust, m_dev_chan, m_pCacher->GetPlayEndTime());
                    m_last_check_cache.resetTime();
                }
            }
            //PrintMemory("缓存的前面字节",pframe->data(),0,128);
            //PrintMemory("缓存的后字节",pframe->data()-8,0,8);
            m_muxer->inputFrame(pframe);
        }
        return true;
    }, m_poller);
    return 0;
}
#endif
template<typename frame>
void FrameRecordCache<frame>::CacheSubScriber::CloseSubscriber() {
    m_send_pkt_timer = nullptr;
    m_stateType = SubscriberStateType::SST_CLOSED;
}

template<typename frame>
bool FrameRecordCache<frame>::CacheSubScriber::IsClosed() {
    return m_stateType == SubscriberStateType::SST_CLOSED;
}

template<typename frame>
std::string FrameRecordCache<frame>::CacheSubScriber::GetNewStreamId2()const {
    return m_dev_chan.GenStreamId2();
}
template<typename frame>
int FrameRecordCache<frame>::CacheSubScriber::GetReaderCount() const {
    return m_muxer->totalReaderCount();
}
//////////////////////////////////////////////FrameRecordCache缓存片段实现/////////////////////////////////////////////////////////////
template<typename frame>
FrameRecordCache<frame>::FrameRecordCache(time_t start_time,DeviceMediaBase *pCache):m_pCacher(pCache),m_ptstimebase_const(1000) {
    m_list_start_time = start_time;
    m_req_start_time = start_time;
    m_first_pts = 0;
    m_last_pts = 0;
    m_cachedmemsize = 0;
    m_visit_time = time(NULL);
    m_abnormal = false;
    m_framecache = std::make_shared<frame_list>();
}
template<typename frame>
FrameRecordCache<frame>::~FrameRecordCache(){
    #ifdef ENABLE_COOLIODEBUG
    /*for(int target_frame_pos = 0; target_frame_pos < (int)m_framecache->size() ; target_frame_pos++) {//从前面往后面查找
        std::shared_ptr<frame> &  pframe = (*m_framecache.get())[target_frame_pos];
        if(target_frame_pos%50 == 0) {
            printf("target_frame_pos=%d的帧的引用计数:%ld,cacheable:%d\n",target_frame_pos,pframe.use_count(),pframe->cacheAble());
        }
    }*/
    #endif
    m_framecache->clear();
    m_list_subscribers.clear();
}
template<typename frame>
void FrameRecordCache<frame>::inputPacket(const std::shared_ptr<frame> &pkt,int pts_offset) {
//     static int counter = 0;
//     if(counter ++ %500 ==0) {
//         printf("[%s:%d]引用计数:%ld,cacheable:%d\n",__FUNCTION__,__LINE__,pkt.use_count(),pkt->cacheAble());
//         clearCache();
//     }
    if(pkt->playStartTime() != m_req_start_time) {
        return;//无效数据输入
    }
    pkt->offset_dts(pts_offset);
    pkt->offset_pts(pts_offset);
    m_visit_time = time(NULL);
    if(pkt->getTrackType() == TrackVideo) {
        if(m_first_pts == 0) {
            m_first_pts = pkt->pts();
        }
        if(privateproto::IsBrokenTimestamp(pkt->pts(),m_last_pts)) {
            if(m_last_pts > pkt->pts()) {
                WarnL << m_pCacher->GetDesc() << "时间戳跳变," << getTimeStr("%Y-%m-%d %H:%M:%S",GetPlayBeginTime()) << "-" << getTimeStr("%Y-%m-%d %H:%M:%S",GetPlayEndTime()) << ",m_last_pts:" << m_last_pts << ",pkt's pts:" << pkt->pts();
                m_abnormal = true;
            }
        }
        if(m_abnormal) {
            return;
        }
        m_last_pts = pkt->pts();
    }
    m_cachedmemsize += pkt->size();
    //追加数据到最后
    m_framecache->emplace_back(pkt);
    #if !defined(NDEBUG)
    //printf("当前缓存大小:%lu,缓存时间长度:%ld毫秒\n",m_framecache->size(),GetTimeLengthMS());
    #endif
} 

template<typename frame>
void FrameRecordCache<frame>::clearCache() {
    m_framecache->clear();
}

template<typename frame>
int64_t FrameRecordCache<frame>::GetTimeLengthMS() {
    return (m_last_pts-m_first_pts);
} 

template<typename frame>
time_t FrameRecordCache<frame>::GetPlayBeginTime() {
    return m_list_start_time;
} 

template<typename frame>
time_t FrameRecordCache<frame>::GetPlayEndTime() {
    return m_list_start_time+(m_last_pts-m_first_pts)/m_ptstimebase_const;
}

template<typename frame>
std::shared_ptr<frame> FrameRecordCache<frame>::GetFrame(size_t frameseq) {
    if(frameseq >= m_framecache->size()) {
        return nullptr;
    }
    auto pframe = (*m_framecache.get())[frameseq];
    return pframe;
}

template<typename frame>
size_t FrameRecordCache<frame>::GetFrameCount() {
    return m_framecache->size();
}

template<typename frame>
int FrameRecordCache<frame>::GetFps() const {
    if(m_framecache->size() == 0 || m_first_pts == 0 || m_last_pts == 0 || m_first_pts == m_last_pts) {
        return 30;//默认值
    }else{
        //return (int)(m_framecache->size()/((m_last_pts-m_first_pts)/m_ptstimebase_const));
        assert(m_last_pts > m_first_pts);
        uint32_t seconds = (m_last_pts-m_first_pts)/m_ptstimebase_const + (((m_last_pts-m_first_pts)%m_ptstimebase_const==0)?0:1);
        int video_frame_count = 0;
        auto it = m_framecache->begin();
        for(;it != m_framecache->end(); it++){
            std::shared_ptr<frame> & pframe = *it;
            if(pframe->getTrackType() == TrackVideo) {
                video_frame_count++;
            }
        }
        return (int)(video_frame_count/seconds);
    }
}

template<typename frame>
int FrameRecordCache<frame>::GetPTSTimeBase() const {
    return m_ptstimebase_const;
}

template<typename frame>
int FrameRecordCache<frame>::LocateFrame(time_t start_push_time,time_t end_push_time){
    uint32_t       target_pts = m_first_pts + (start_push_time-m_list_start_time)*m_ptstimebase_const;
    int        target_frame_pos = (int)((start_push_time-m_list_start_time)*GetFps());
    Ticker     ticker;
    m_visit_time = time(NULL);
    #if !defined(NDEBUG)  
    TraceL << "起始pts:" << m_first_pts << ",目标pts:" << target_pts << ",[" << m_list_start_time << "-" << start_push_time << "] target_frame_pos:" << target_frame_pos;
    #endif
    size_t cur_cache_size = m_framecache->size();
    if(cur_cache_size == 0) {
        return -1;
    }
    if(target_frame_pos >= (int)cur_cache_size) {
        target_frame_pos = (int)cur_cache_size -1;
    }
    
    int tmp_curPushPosInList = 0;
    auto  pframe = (*m_framecache.get())[target_frame_pos];
    if(pframe->pts() < target_pts) {
        #if !defined(NDEBUG)
        printf("[%s:%d]起始pts:%u,目标pts:%u,pframe->pts:%u,target_pts:%u\n",__FUNCTION__,__LINE__,m_first_pts,target_pts,pframe->pts(),target_pts);
        #endif
        for(; target_frame_pos < (int)cur_cache_size ; target_frame_pos++) {//从前面往后面查找
            auto  pframe = (*m_framecache.get())[target_frame_pos];
             #ifdef ENABLE_COOLIODEBUG 
            if(pframe->configFrame()||pframe->keyFrame()) {
            #else
            if(!pframe->configFrame()) {
            #endif
                continue;
            }
            if(pframe->pts() > target_pts) {
                tmp_curPushPosInList = target_frame_pos;
                #if !defined(NDEBUG)
                printf("[%s:%d]找到目标pts:%u的关键帧%u,[%lu-%lu],tmp_curPushPosInList:%d,target_frame_pos:%u\n",__FUNCTION__,__LINE__,target_pts,pframe->pts(),m_list_start_time,start_push_time,tmp_curPushPosInList,target_frame_pos);
                #endif
                break;
            }
             #if !defined(NDEBUG)
            /*else{         
                printf("[%s:%d]正在找目标pts:%u的关键帧%u,[%lu-%lu],target_frame_pos:%u\n",__FUNCTION__,__LINE__,target_pts,pframe->pts(),m_list_start_time,start_push_time,target_frame_pos);
            }*/
            #endif
        }
    }else {
        #if !defined(NDEBUG)
        printf("[%s:%d]起始pts:%u,目标pts:%u,pframe->pts:%u,target_pts:%u\n",__FUNCTION__,__LINE__,m_first_pts,target_pts,pframe->pts(),target_pts);
        #endif
        for(; target_frame_pos >= 0 ; target_frame_pos--) {//从后面往前面查找
            auto  pframe = (*m_framecache.get())[target_frame_pos];
             #ifdef ENABLE_COOLIODEBUG 
            if(pframe->configFrame()||pframe->keyFrame()) {
            #else
            if(!pframe->configFrame()) {
            #endif
                continue;
            }
            if(pframe->pts() < target_pts) {
                tmp_curPushPosInList = target_frame_pos;
                #if !defined(NDEBUG)
                printf("[%s:%d]找到目标pts:%u的关键帧%u,[%lu-%lu],tmp_curPushPosInList:%d\n",__FUNCTION__,__LINE__,target_pts,pframe->pts(),m_list_start_time,start_push_time,tmp_curPushPosInList);
                #endif
                break;
            }
            #if !defined(NDEBUG)
            /*else{          
                printf("[%s:%d]正在找目标pts:%u的关键帧%u,[%lu-%lu],target_frame_pos:%u\n",__FUNCTION__,__LINE__,target_pts,pframe->pts(),m_list_start_time,start_push_time,target_frame_pos);
            }*/
            #endif
        }
    }
    TraceL << "找到[" << start_push_time << "-" << end_push_time << "] 返回帧序号:" << tmp_curPushPosInList << "查找消耗时间:" << ticker.elapsedTime() << "毫秒";
    return tmp_curPushPosInList;
}

template<typename frame>
Value FrameRecordCache<frame>::DumpAsJson() {
    Value root;
    root["req_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",m_req_start_time);
    root["list_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",GetPlayBeginTime());
    root["list_end_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",GetPlayEndTime());
    root["last_visit_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",m_visit_time);
    root["first_pts"] = Int64(m_first_pts);
    root["last_pts"] = Int64(m_last_pts);
    root["ptstimebase"] = Int64(m_ptstimebase_const);
    root["TimeLengthMS"] = Int64(GetTimeLengthMS());
    root["frame_count"] = Int64(m_framecache->size());
    root["fps"] = Int64(GetFps());
    root["FrameCacheMemorySize"] = Int64(FrameCacheMemorySize());
    Json::Value subcribers(Json::arrayValue);
    auto it = m_list_subscribers.begin();
    for(;it != m_list_subscribers.end();it++) {
        subcribers.append((*it)->DumpAsJson());
    }
    root["subcribers"] = subcribers;
    return root;
}
template<typename frame>
int64_t FrameRecordCache<frame>::PurgeMemory(const int machine_memory,const int process_memory){
    if(m_list_subscribers.size() <=0 && (IsAbnormal()||(time(NULL) - m_visit_time) > MAX_CACHE_IDLE_TIME)) {
        WarnL << "缓存片段["<< getTimeStr("%Y-%m-%d %H:%M:%S",GetPlayBeginTime()) << "-" << getTimeStr("%Y-%m-%d %H:%M:%S",GetPlayEndTime())<< "]闲置超过一小时,删除缓存";
        return (0-FrameCacheMemorySize());//准备删除
    }else{
        return 0;//保留
    }
}
template<typename frame>
int64_t FrameRecordCache<frame>::FrameCacheMemorySize(){
#if 0
    int64_t cache_memory = 0;
    auto it = m_framecache->begin();
    for(;it != m_framecache->end();it++){
        std::shared_ptr<frame> & pframe = *it;
        //printf("[%s:%d]引用计数:%ld\n",__FUNCTION__,__LINE__,pframe.use_count());
        cache_memory += int64_t(pframe->getCapacity());        
    }
    return cache_memory;
#else
    return m_cachedmemsize;
#endif
}
template<typename frame>
bool FrameRecordCache<frame>::IsAbnormal() const {
    return m_abnormal;
}
DeviceRecordCache::DeviceRecordCache(const privateproto::DeviceChannel& dev_chan, bool autosubscribe):m_haveVideo(false),m_haveAudio(false),m_autosubscribe(autosubscribe),m_dev_chan(dev_chan),m_visit_time(time(NULL)) {
    m_check_merge_timer = std::make_shared<Timer>(5,[=](){
        CheckRecordCacheMerge();
        return true;
    }, m_poller);
}

DeviceRecordCache::~DeviceRecordCache() {
    m_record_cache_container.clear();
    m_check_merge_timer = nullptr;
}

void DeviceRecordCache::inputFrame(const Frame::Ptr &frame) {
    #if !defined(NDEBUG)
    /*if(frame->keyFrame()||frame->configFrame()){
        printf("[%s:%s:%d.%s]传入帧的播放开始时间:%ld,keyFrame:%d,configFrame:%d,pts:%u,%x|%x|%x|%x|%x\n",__FILE__,__FUNCTION__,__LINE__,getTimeStr("%Y-%m-%d %H:%M:%S",time(NULL)).c_str(),frame->playStartTime(),frame->keyFrame(),frame->configFrame(),frame->pts(),frame->data()[0],frame->data()[1],frame->data()[2],frame->data()[3],frame->data()[4]);
    }*/
    #endif
    m_poller->async([this,frame]() {
        TimeTicker1(5);
        m_visit_time = time(NULL);
        time_t start_time = frame->playStartTime();   
        int pts_offset = 0;
        auto find_tie = m_cachetier.find(start_time);
        if(find_tie != m_cachetier.end()) {
            start_time = find_tie->second.target_start_time;
            pts_offset = find_tie->second.pts_offset;
            //DebugL << getTimeStr("%Y-%m-%d %H:%M:%S",find_tie->first) << "转向" << getTimeStr("%Y-%m-%d %H:%M:%S",find_tie->second.target_start_time) << "提交缓存";
        }
        auto itfind = m_record_cache_container.find(start_time);
        if ( itfind == m_record_cache_container.end()) {
            FrameRecordCache<Frame>::Ptr record = std::make_shared<FrameRecordCache<Frame>>(start_time,this);
            record->inputPacket(frame,pts_offset);
            if(m_autosubscribe){
                InfoL << "自动创建订阅者" << m_dev_chan.ToString() << ",start_time:" << getTimeStr("%Y-%m-%d %H:%M:%S",start_time);
                std::string stream_id = m_dev_chan.GenStreamId();
                record->NewCacheSubScriber(m_poller,m_dev_chan,m_tracks,start_time,start_time,stream_id)->StartPushRecordStream();
            }
            m_record_cache_container[start_time] = record;
        }else {
            itfind->second->inputPacket(frame,pts_offset);
        }
    });
}

bool DeviceRecordCache::CheckRecordCacheMerge() {
    for(auto it = m_record_cache_container.begin();it != m_record_cache_container.end();it++) {
        FrameRecordCache<Frame>::Ptr & cacher =  it->second;
        cacher->CheckCacheSubScriber();
    }
    while(true) {
        time_t tmp_merged_time = 0;
        bool domerge = false;
        auto it = m_record_cache_container.begin();
        auto it_prev = it;
        for(;it != m_record_cache_container.end();it++) {
            if(tmp_merged_time == 0) {
                tmp_merged_time = it->first + it->second->GetTimeLengthMS()/1000;
                //DebugL << "第一片缓存:" << it->second->DumpAsJson().toStyledString();
            }else if(tmp_merged_time >= it->first) {
                DebugL << "需要做合并操作," << it_prev->second->DumpAsJson().toStyledString() << "|" << it->second->DumpAsJson().toStyledString();
                //需要合并
                int pts_offset = it_prev->second->MergeOtherFrameRecordCache(it->first,*(it->second.get()));
                domerge = true;
                UpdateCacheTier(it->first,it_prev->first,pts_offset);
                m_record_cache_container.erase(it);
                break;
            }/*else {
                DebugL << "不需要做合并操作," << it_prev->second->DumpAsJson().toStyledString() << "|" << it->second->DumpAsJson().toStyledString();
            }*/
            it_prev = it;
        }
        if(!domerge) {
            break;
        }
    }
    return true;
}

void DeviceRecordCache::play_record_resources(privateproto::DeviceChannel& dev_chan,time_t startTime,time_t endTime,std::function<void(bool ,const std::string,const std::string )> cb) {
    TraceL << "播放缓存资源, startTime:" << startTime << ",endTime:" << endTime << ",dev_chan:" << dev_chan.ToString();
    m_poller->async([this,dev_chan,startTime,endTime,cb]() {
        TraceL << "开始执行async, startTime:" << startTime << ",endTime:" << endTime << ",dev_chan:" << dev_chan.ToString();
        m_visit_time = time(NULL);
        std::vector<TimeRange> resources_container;
        std::string new_stream_id;
        struct tm* ptmStart = gmtime(&startTime);    
        auto it = m_record_cache_container.begin();
        for(;it != m_record_cache_container.end();it++) {
            FrameRecordCache<Frame>::Ptr & cacher =  it->second;
            if(cacher->IsAbnormal()||cacher->GetTimeLengthMS() < 10000)
                continue;
            time_t cur_start = cacher->GetPlayBeginTime();
            struct tm* ptmStartIt = gmtime(&cur_start);
            if(ptmStart->tm_year == ptmStartIt->tm_year &&  ptmStart->tm_mon == ptmStartIt->tm_mon &&  ptmStart->tm_mday == ptmStartIt->tm_mday && startTime>=cacher->GetPlayBeginTime()) {//同一天
                TimeRange range(FFMAX(startTime,cacher->GetPlayBeginTime()),FFMIN(endTime,cacher->GetPlayEndTime()));
                if(range.IsValid()) {
                    resources_container.push_back(range);
                    TraceL << "找到缓存,开始发送流, startTime:" << startTime << ",endTime:" << endTime;
                    cacher->NewCacheSubScriber(m_poller,dev_chan,m_tracks,startTime,endTime,new_stream_id)->StartPushRecordStream();
                    break;
                }
            }
        }
        cb(resources_container.size() > 0,dev_chan.GenStreamId() ,new_stream_id);
    });
}

void DeviceRecordCache::control_cache_stream(privateproto::DeviceChannel& dev_chan, const std::string command, const double scale,std::function<void(bool controlsuccess)> cb){
    TraceL << "控制缓存流, command:" << command << ",scale:" << scale;
    m_poller->async([this,dev_chan,command,scale,cb]() {
            TraceL << "开始执行async, command:" << command << ",scale:" << scale;
            m_visit_time = time(NULL);
            auto it = m_record_cache_container.begin();
            for(;it != m_record_cache_container.end();it++) {
                FrameRecordCache<Frame>::Ptr & cacher =  it->second;
                auto subcriber = cacher->FindCacheSubScriber(dev_chan);
                if(subcriber) {
                    subcriber->ControlStream(command,scale);
                    cb(true);
                    return;
                }
            }
            cb(false);
    });
}

void DeviceRecordCache::close_cache_stream_subscriber(privateproto::DeviceChannel& dev_chan) {
    m_poller->async([this,dev_chan]() {
        auto it = m_record_cache_container.begin();
        for(;it != m_record_cache_container.end();it++) {
            FrameRecordCache<Frame>::Ptr & cacher =  it->second;
            cacher->CloseCacheSubScriber(dev_chan);
        }
    });
}
void DeviceRecordCache::addTrack(const Track::Ptr & track) {
    printf("[%s:%s:%d]添加track\n",__FILE__,__FUNCTION__,__LINE__);
    m_tracks.emplace_back(track);
    if(track->getTrackType() == TrackVideo){
        m_haveVideo = true;
    }
    if(track->getTrackType() == TrackAudio){
        m_haveAudio = true;
    }
}

void DeviceRecordCache::addTrackCompleted() {
    printf("[%s:%s:%d]添加track完成\n",__FILE__,__FUNCTION__,__LINE__);
}

void DeviceRecordCache::resetTracks() {
    printf("[%s:%s:%d]重置track\n",__FILE__,__FUNCTION__,__LINE__);
    m_tracks.clear();
    m_haveVideo = false;
    m_haveAudio = false;
}

std::map<time_t, FrameRecordCache<Frame>::Ptr> DeviceRecordCache::getCacheContainer() {
    return m_record_cache_container;
}
void DeviceRecordCache::DumpAsJsonInfo(const function<void(const Json::Value& info)> &cb){
    m_poller->async([this,cb]() {
        try {
            Json::Value root;
            root["hasVideo"] = m_haveVideo;
            root["haveAudio"] = m_haveAudio;
            Json::Value list(Json::arrayValue);
            
            auto it = m_record_cache_container.begin();
            for(;it != m_record_cache_container.end();it++) {
                list.append(it->second->DumpAsJson());
            }
            root["cachelist"] = list;
            Json::Value ties(Json::arrayValue);
            auto it2 = m_cachetier.begin();
            for(;it2 != m_cachetier.end();it2++) {
                Json::Value item;
                item["after_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",it2->first);
                item["before_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",it2->second.target_start_time);
                ties.append(item);
            }
            root["ties"] = ties;
            //JSON_ASSERT_MESSAGE(0,"测试json异常");//行为符合预期
            cb(root);
        }catch (std::exception &ex) {
            WarnL << ex.what();
            System::printcallstack("捕获Json异常");
        
            Json::Value root;
            root["hasVideo"] = m_haveVideo;
            root["haveAudio"] = m_haveAudio;
            root["info"] = ex.what();
            cb(root);
        }catch(...) {
            WarnL << "Unknown error";
            System::printcallstack("捕获Json异常");
            Json::Value root;
            root["hasVideo"] = m_haveVideo;
            root["haveAudio"] = m_haveAudio;
            root["info"] = "Unknown error";
            cb(root);
        }
    });
}
Value DeviceRecordCache::DumpAsJson() {
    Json::Value root;
    std::promise<bool> prom;
    std::future<bool> fut = prom.get_future();
    m_poller->async([this,&root,&prom]() {
        try {
            root["hasVideo"] = m_haveVideo;
            root["haveAudio"] = m_haveAudio;
            Json::Value list(Json::arrayValue);
            
            auto it = m_record_cache_container.begin();
            for(;it != m_record_cache_container.end();it++) {
                list.append(it->second->DumpAsJson());
            }
            root["cachelist"] = list;
            
            Json::Value ties(Json::arrayValue);
            auto it2 = m_cachetier.begin();
            for(;it2 != m_cachetier.end();it2++) {
                Json::Value item;
                item["after_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",it2->first);
                item["before_start_time"] = getTimeStr("%Y-%m-%d %H:%M:%S",it2->second.target_start_time);
                ties.append(item);
            }
            root["ties"] = ties;
            prom.set_value(true);
        }catch (std::exception &ex) {
            WarnL << ex.what();
            System::printcallstack("捕获Json异常");
            prom.set_value(true);
        }catch(...) {
            WarnL << "Unknown error";
            System::printcallstack("捕获Json异常");
            prom.set_value(true);
        }
    });
    fut.get();
    return root;
}
std::string DeviceRecordCache::GetDesc() {
    return m_dev_chan.GenStreamId();
}
int64_t DeviceRecordCache::DeviceCacheMemorySize(){
    int cached_memory = 0;
    std::promise<bool> prom;
    std::future<bool> fut = prom.get_future();
    m_poller->async([this,&cached_memory,&prom]() {
        auto it = m_record_cache_container.begin();
        for(;it != m_record_cache_container.end();it++) {
            cached_memory += it->second->FrameCacheMemorySize();
        }
        prom.set_value(true);
    });
    fut.get();
    return cached_memory;
}
int DeviceRecordCache::PurgeMemory(const int machine_memory,const int process_memory) {
    int erased_memory = 0;
    std::promise<bool> prom;
    std::future<bool> fut = prom.get_future();
    m_poller->async([this,machine_memory,process_memory,&erased_memory,&prom]() {
        TimeTicker1(5);
        if(m_record_cache_container.size() <= 0 && (time(NULL) - m_visit_time) > MAX_CACHE_IDLE_TIME) {
            WarnL << "设备缓存" << m_dev_chan.ToString() << "闲置超过一小时,删除缓存";
            erased_memory = (0-int(sizeof(DeviceRecordCache)));
        }else{
            auto it = m_record_cache_container.begin();
            for(;it != m_record_cache_container.end();) {
                /*if(it->second->IsAbnormal()){//让它超时再删除,不要突然一下子就删除
                    it = m_record_cache_container.erase(it);
                }else*/{
                    int cur_erased_memory = it->second->PurgeMemory(machine_memory,process_memory);
                    if(cur_erased_memory < 0) {
                        erased_memory -= cur_erased_memory;
                        WarnL << "设备缓存片段删除";
                        //先维护m_cachetier
                        for(auto it_tie = m_cachetier.begin();it_tie != m_cachetier.end();) {
                            if(it_tie->second.target_start_time == it->first){
                                m_cachetier.erase(it_tie++);
                            }else{
                                it_tie++;
                            }
                        }
                        m_record_cache_container.erase(it++);
                    }else {
                        erased_memory += cur_erased_memory;
                        it++;
                    }
                }
            }
        }
        prom.set_value(true);
    });
    fut.get();
    return erased_memory;
}
void DeviceRecordCache::CleanCache(){
    std::promise<bool> prom;
    std::future<bool> fut = prom.get_future();
    m_poller->async([this,&prom]() {
        auto it = m_record_cache_container.begin();
        for(;it != m_record_cache_container.end();it++) {
            it->second->clearCache();
        }
        m_record_cache_container.clear();
        prom.set_value(true);
    });
    fut.get();
}

bool DeviceRecordCache::isEnabled() {
    return true;
}
int DeviceRecordCache::totalReaderCount() const{
    return 1;
}
